<?php
/***************************************************************************
 * 
 * Copyright (c) 2011 , Inc. All Rights Reserved
 * $Id$:GDP.php,2011/06/27 19:59:55 
 * 
 **************************************************************************/
 
 
 
/**
 * @file GDP.php
 * @author huqingping
 * @date 2011/06/27 19:59:55
 * @version 1.0 
 * @brief 
 *  
 **/

require_once 'rpc/RpcFactory.php';

final class GDP 
{
	const SVR_NAME = 'gdp';
	const HITS_USE = 0x02;
	const HITS_MASTER = 0x01;

	//rpc handle
	private $rpc;

	private $conf;
	private $dbname;

	private $forceUse = false;
	private $forceMaster = false;

	//事务
	private $tx = false;
	private $txid = 0;
	//Phpproxy的接口，主要为了实现事务连接保持功能
	private $shardId = 0;
	private $shardCount = 0;

	function __construct($db,$conf) 
	{
		$this->dbname = $db;
		$this->conf = $conf;
		$this->forceUse = !empty($conf['force_use']);
		unset($this->conf['force_use']);
	}

	private function initRpc()
	{
		$svrname = self::SVR_NAME;
		if (!$this->rpc) {
			//延迟创建rpc对象
			$this->rpc = RpcFactory::getRpc(false);
			$this->rpc->init($svrname,$this->conf);
		}
	}

	function forceMaster($force=true)
	{
		$this->forceMaster = $force;
	}
	function forceUse($force=true)
	{
		$this->forceUse = $force;
	}

	/**
	 * 从gdp获得一个全局id
	 * @params count 表示分配多少个id，如果分配多个id，只返回最后一个id
	 */ 
	function guid($name,$count=1)
	{
		$count = intval($count);
		if ($count < 1) {
			throw new Exception("db.guid error count");
		}
		$req = array('name'=>$name,'count'=>$count);
		if (DbContext::$logFunc) {
			$log = '[GUID][NAME:'.$name.'][COUNT:'.$count.']';
			call_user_func(DbContext::$logFunc,$log);
		}
		if (DbContext::$testMode) {
			return 1;
		}
		$this->initRpc();
		$ret = $this->rpc->rpcCall('guid',$req);
		if ('ok' != $ret['err']) {
			throw new Exception('db.GUIDError '.$ret['err'].' '.$ret['err_detail']);
		}
		return $ret['id'];
	}

	/**
	 * 执行一个查询
	 * 参数为执行查询需要的sql 值等信息，格式如下:
	 * query('select');
	 * query('select', $arg,$arg);
	 * query(array(
	 *	array('select')
	 *));
	 * query(array(
	 *		array('select1', arg,$arg),
	 *		array('select2', arg,$arg),
	 *	));
	 */ 
	function query($sqlInfo)
	{
		if (empty($sqlInfo)) {
			throw new Exception('db.NotAllowEmptyQuery');
		}
		if ((isset(TxScope::$txDbs[$this->dbname]) || TxScope::$tx) && !$this->tx) {
			//如果开启了全局事务并且当前还没有执行过事务
			$this->beginTx();
		}
		$sqls = $this->buildQuery($sqlInfo);
		$req = array('db'=>$this->dbname,'sqls'=>$sqls,'hints'=>0);
		if ($this->tx) {
			if ($this->txid) {
				//如果开启事务了并且已经拿到了事务id，就需要带上事务id
				$req['txid'] = $this->txid;
			}
			if ($this->shardCount > 0) {
				//需要保持事务连接
				$req['_RPC_SHARD_ID'] = $this->shardId;
				$req['_RPC_SHARD_COUNT'] = $this->shardCount;
			}
		}
		if ($this->forceUse) {
			//兼容模式，
			$req['hints'] = $req['hints'] | self::HITS_USE;
		}	
		if ($this->forceMaster) {
			//兼容模式，
			$req['hints'] = $req['hints'] | self::HITS_MASTER;
		}	
		//LOG SQL
		if (DbContext::$logFunc) {
			$log = '[TXID:'.$this->txid.'][DB:'.$this->dbname.'][HINTS:'.$req['hints'].'][SQL:'.implode(';',$sqls).']';
			call_user_func(DbContext::$logFunc,$log);
		}
		if (DbContext::$testMode) {
			//测试模式，直接返回空结果
			return array(
				'err'=>'ok',
				'results'=>array(
					array('fields'=>array(),'affected_rows'=>0,'found_rows'=>0)
				),
				'txid'=>1232
			);
		}
		$this->initRpc();
		$ret = $this->rpc->rpcCall('query',$req);
		if ('ok' != $ret['err']) {
			$this->txid = 0;
			$this->tx = false;
			$msg = isset($ret['err_detail'])?$ret['err_detail']:'';
			throw new Exception('db.QueryError '.$ret['err'].' '.$msg);
		}
		if ($this->tx) {
			//开启了事务
			if (!$this->txid) {
				$this->txid = $ret['txid'];
			}
			$this->shardId = $ret['_RPC_SHARD_ID'];
			$this->shardCount = $ret['_RPC_SHARD_COUNT'];
		}
		return $ret;
	}

	private function buildQuery($queryInfo)
	{
		if (is_string($queryInfo)) {
			$querys = array(array($queryInfo));
		} elseif (is_array($queryInfo[0])) {
			$querys = $queryInfo[0];
		} else {
			$querys = array($queryInfo);
		}
		$ret = array();
		foreach($querys as $query) {
			$sql = array_shift($query);
			if (isset($query[0]) && is_array($query[0])) {
				//如果第一个是数组，则认为所有的参数都在这个数组里面
				$query = $query[0];
			}
			$argnum = count ($query);
			if (substr_count ( $sql, '?' ) >  $argnum) {
				throw new Exception ( "db.MysqlSqlParam:$sql Error" );
			}
			if ($argnum > 0) {
				$format_sql = str_replace ( '?', '%s', $sql);
				$ret[] = vsprintf($format_sql, $this->escapeValues($query));
			} else {
				$ret[] = $sql;
			}
		}
		return $ret;
	}

	function escapeValues($arr)
	{
		foreach($arr as &$v) {
			$v = $this->escapeValue($v);
		}
		return $arr;
	}

	function escapeValue($value) {
		if( is_string($value) ){
			$hex_value = bin2hex( $value );
			return "unhex('$hex_value')";
		} elseif (is_numeric( $value)) {
			if (0 == $value) {
				return "'0'";
			}
			return $value;
		}elseif ( is_null($value) ){
			return 'NULL';
		}elseif (is_bool($value)) {
			//布尔值返回1/0
			return $value?1:"'0'";
		} elseif (is_array($value) && isset($value['exp']) && is_string($value['exp'])) {
			//支持字段值为exp表达式
			return $value['exp'];
		} else {
			throw new Exception('db.EscapeValue not support type');
		}
	}

	function beginTx()
	{
		if ($this->tx || $this->txid) {
			throw new Exception('db.Transaction already begins');
		}
		$this->tx = true;
		try {
			return $this->query('START TRANSACTION');
		}catch(Exception $ex) {
			if (!$this->txid) {
				//如果txid没分配成功，直接设置tx=false表示事务没启动成功
				$this->tx = false;
			}
			throw $ex;
		}
	}

	private function stopTrans($sql)
	{
		if (!$this->tx || !$this->txid) {
			return;
		}

		$ret = $this->query($sql);
		$this->tx = false;
		$this->txid = 0;
		$this->shardId = 0;
		$this->shardCount = 0;
		return $ret;
	}

	function commit()
	{
		return $this->stopTrans('COMMIT');
	}

	function rollback()
	{
		return $this->stopTrans('ROLLBACK');
	}

	function isTxBegin()
	{
		return $this->tx;
	}
}




/* vim: set ts=4 sw=4 sts=4 tw=100 noet: */
